package com.jqj.example;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.util.context.Context;

import java.io.IOException;
import java.time.Duration;
import java.time.LocalTime;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @Desc: Reactive Stream
 * @Author: lijingzheng
 * @Date: 2025/1/6
 */
public class ReactiveStreamTest {

    public static void main(String[] args) throws Exception {

        // 创建一个简单的发布者
//        Flux<String> publisher = Flux.just("Hello", "World", "Reactive", "Streams");

//        Flux<String> publisher = Flux.fromIterable(Arrays.asList("Hello-Reactive", "World-Streams"))
//                .flatMap(s -> Flux.fromArray(s.split("-")))
//                .distinct().zipWith(Flux.range(1, Integer.MAX_VALUE),
//                        (string, count) -> String.format("%s-%2d", string, count));

//        Flux<String> publisher = Flux.interval(Duration.ofSeconds(10), Duration.ofSeconds(1))
//               .map(i -> "Hello-" + i).take(10);

//        Flux<String> publisher = Mono.just("Hello")
//                .concatWith(Mono.just("World"))
//                .delayUntil((Function<String, Mono<Long>>) s -> {
//                    if (s.equals("Hello")) {
//                        return Mono.delay(Duration.ofSeconds(3));
//                    }
//                    if (s.equals("World")) {
//                        return Mono.delay(Duration.ofSeconds(5));
//                    }
////                    return Mono.just(s + "-" + LocalTime.now());
//                    return Mono.empty();
//                });

//        Flux<String> publisher = Mono.just("Hello")
//                .delayUntil((Function<String, Mono<Long>>) s -> {
////                    return Mono.just(s + "-" + LocalTime.now())
//                    return Mono.delay(Duration.ofSeconds(3));
//                })
//                .concatWith(Mono.just("World"))
//                .delayUntil((Function<String, Mono<Long>>) s -> {
////                    return Mono.just(s + "-" + LocalTime.now())
//                    // Hello delay 3+5=8s, World delay 5s
//                    return Mono.delay(Duration.ofSeconds(5));
//                });

//        Mono<String> mono = Mono.just("oops I'm late")
//                .delaySubscription(Duration.ofSeconds(5)); // 延迟订阅时间，影响订阅者何时开始接收发布者的信号
//        Flux<String> flux = Flux.just("let's get", "the party", "started")
//                .delaySequence(Duration.ofSeconds(2)) // 延迟序列发送时间，影响发布者何时开始发送序列中的元素
//                .delayElements(Duration.ofSeconds(1)); // 延时数据流中的每个元素
//
//        // 选择发送信号的第一个 publisher
//        Flux<String> publisher = Flux.firstWithSignal(mono, flux)
//                .delaySubscription(Duration.ofSeconds(5))
//                .delayElements(Duration.ofSeconds(1)); // 此处的 element 就一个，是指 flux 或 mono 的整个数据流
//
//        Flux<String> publisher = Flux.combineLatest(Flux.just("Hello", "Word", "X"),
//                Flux.just(1), (x, y) -> x + "-" + y);
//        Flux<String> publisher = Flux.combineLatest(xArr -> Arrays.stream(xArr).map(Object::toString).collect(Collectors.joining(" -> ")),
//                Flux.just("Hello", "Word").delayUntil(x -> Mono.delay(Duration.ofMillis(new Random().nextInt(1000)))),
//                Flux.just(1, 2).delayUntil(i -> Mono.delay(Duration.ofSeconds(i * 10))),
//                Flux.just('A', 'B', 'C', 'D')); // FA,FB,FC 三个 Flux，比如 FB 中元素最晚发布，
        // 那组合发布的时间就取决于 FB 中每个元素的发布时间，当 FB 中的第一个元素 FB-1 发布时，组合 FA 和 FC 最新发布的元素

//        AtomicReference<Consumer<String>> producer = new AtomicReference<>();
//        AtomicReference<Supplier<String>> action = new AtomicReference<>();
//        Flux<String> publisher = Flux.create(sink -> {
//            producer.set(sink::next);
//            action.set(() -> {
//                sink.complete();
//                return "Completed";
//            });
//        }, FluxSink.OverflowStrategy.LATEST);
//
//        publisher.subscribe(
//                item -> {
//                    if (item.equals("Word")) {
//                        try {
//                            TimeUnit.SECONDS.sleep(5);
//                        } catch (InterruptedException e) {
//                            throw new RuntimeException(e);
//                        }
//                    }
//                    System.out.printf("\033[34;1m%s ===>> %s ===>> %s\033[0m\n",
//                            LocalTime.now(), Thread.currentThread().getName(), "Received: " + item);
//                },
//                error -> System.out.printf("\033[35;1m%s ===>> %s ===>> %s\033[0m\n",
//                        LocalTime.now(), Thread.currentThread().getName(), "Error: " + error),
//                () -> System.out.printf("\033[36;1m%s ===>> %s ===>> %s\033[0m\n",
//                        LocalTime.now(), Thread.currentThread().getName(), "Completed"));

//        // 创建一个 Supplier，每次调用时返回一个新的 Flux
//        Supplier<Flux<Integer>> fluxSupplier = () -> Flux.range(1, 5);
//        // 使用 Flux.defer 创建一个延迟创建的 Flux
//        Flux<Integer> deferredFlux = Flux.defer(fluxSupplier);
//        // 订阅两次，每次都会调用 fluxSupplier 获取一个新的 Flux
//        deferredFlux.subscribe(System.out::println);
//        deferredFlux.subscribe(System.out::println);

        Flux<String> flux = Flux.deferContextual(contextView -> {
            String key = contextView.getOrDefault("key", "default value");
            assert key != null;
            return Flux.just(key);
        });
        // 设置 Context
        Context context = Context.of("key", "custom value");
        // 订阅 Flux，并传递 Context
        flux.contextWrite(context)
                .subscribe(System.out::println);
        flux.contextWrite(context)
                .subscribe(System.out::println);
        // 设置 Context
        Context context2 = Context.of("key", "custom value-2");
        // 订阅 Flux，并传递 Context
        flux.contextWrite(context2)
                .subscribe(System.out::println);


        System.out.printf("\033[33;1m%s === 创建订阅者\033[0m\n", LocalTime.now());

//        // 创建一个订阅者
//        publisher.subscribeOn(Schedulers.boundedElastic())
//                .doOnSubscribe(subscription -> System.out.printf("\033[32;1m%s ===>> %s ===>> %s\033[0m\n",
//                        LocalTime.now(), Thread.currentThread().getName(), "Subscribed"))
//                .doOnNext(item -> System.out.printf("\033[34;1m%s ===>> %s ===>> %s\033[0m\n",
//                        LocalTime.now(), Thread.currentThread().getName(), "Received: " + item))
//                .doOnError(error -> System.out.printf("\033[35;1m%s ===>> %s ===>> %s\033[0m\n",
//                        LocalTime.now(), Thread.currentThread().getName(), "Error: " + error))
//                .doOnComplete(() -> System.out.printf("\033[36;1m%s ===>> %s ===>> %s\033[0m\n",
//                        LocalTime.now(), Thread.currentThread().getName(), "Completed"))
//                .subscribe();


        // 订阅处理为异步线程时需要等待一会
//        TimeUnit.SECONDS.sleep(1);
//        producer.get().accept("Hello");
//        producer.get().accept("Word");
//        action.get().get();
//        producer.get().accept("Word");

        System.in.read();
    }

    public static void main2(String[] args) throws IOException {
        Flux<Integer> publisher = Flux.range(1, 10);

        publisher.subscribe(new Subscriber<Integer>() {

            private Subscription subscription;
            private final AtomicInteger count = new AtomicInteger(0);
            private final AtomicInteger time = new AtomicInteger(0);
            private int requestSize = 3;

            @Override
            public void onSubscribe(Subscription subscription) {
                println("Subscribed: " + subscription);
                this.subscription = subscription;
                this.subscription.request(requestSize);
            }

            @Override
            public void onNext(Integer i) {
                CompletableFuture.runAsync(() -> {
                    println("Received: " + i);

                    try {
                        TimeUnit.SECONDS.sleep(i);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    time.getAndAdd(i);
                    if (time.get() == 10 || time.get() == 20) --requestSize;
                    if (count.incrementAndGet() == requestSize) {
                        count.set(0);
                        subscription.request(requestSize);
                    }
                });
            }

            @Override
            public void onError(Throwable t) {
                println("Error: " + t);
            }

            @Override
            public void onComplete() {
                println("Completed");
            }
        });

        // 延迟以确保异步操作完成
        System.in.read();
    }

    public static void println(String str) {
        System.out.printf("\033[34;1m%s ===>> %s ===>> %s\033[0m\n",
                LocalTime.now(), Thread.currentThread().getName(), str);
    }

}
